package com.cryptoneedle.garden.service.garden.source.metadata;


import com.bubbles.engine.data.core.repository.BaseRepository;
import com.bubbles.engine.data.core.service.BaseService;
import com.cryptoneedle.garden.entity.source.metadata.SourceCatalog;
import com.cryptoneedle.garden.entity.source.metadata.SourceColumn;
import com.cryptoneedle.garden.entity.source.metadata.SourceDatabase;
import com.cryptoneedle.garden.entity.source.metadata.SourceTable;
import com.cryptoneedle.garden.key.ColumnKey;
import com.cryptoneedle.garden.repository.source.metadata.SourceColumnRepository;
import com.cryptoneedle.garden.service.crud.garden.source.*;
import com.cryptoneedle.garden.service.util.JdbcUtil;
import com.google.common.collect.Maps;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>description: 源数据源-列 </p>
 *
 * @author CryptoNeedle
 * @date 2025-06-20
 */
@Slf4j
@Service
public class ColumnService extends BaseService<SourceColumn, ColumnKey> {

    @Resource
    private ColumnService columnService;
    @Resource
    private SourceColumnRepository sourceColumnRepository;

    @Override
    public BaseRepository<SourceColumn, ColumnKey> getRepository() {
        return this.sourceColumnRepository;
    }

    @Autowired
    private FoundSourceService select;
    @Autowired
    private AddSourceService add;
    @Autowired
    private PatchSourceService patch;
    @Autowired
    private SaveSourceService save;
    @Autowired
    private DeleteSourceService delete;

    @Transactional
    public void syncDatabase(SourceCatalog catalog, SourceDatabase database) {
        List<SourceTable> tables = select.tables(database);
        Map<String, SourceTable> tableMap = Maps.uniqueIndex(tables, SourceTable::getTableName);

        JdbcTemplate jdbcTemplate = JdbcUtil.getJdbcTemplate(catalog);
        List<SourceColumn> columns = jdbcTemplate.query(
                catalog.getDatabaseType().getDatabase().getColumnSql(database.getDatabaseName()),
                (resultSet, rowNum) -> {
                    String tableName = resultSet.getString("tableName");
                    SourceTable table = tableMap.get(tableName);
                    SourceColumn column = new SourceColumn()
                            .setCatalogName(database.getCatalogName())
                            .setDatabaseName(database.getDatabaseName())
                            .setTableName(tableName)
                            .setColumnName(resultSet.getString("columnName"))
                            .setComment(resultSet.getString("columnComment"))
                            .setPosition(resultSet.getLong("position"))
                            .setDataType(resultSet.getString("dataType"))
                            .setLength(resultSet.getLong("length"))
                            .setPrecision(resultSet.getLong("precision"))
                            .setScale(resultSet.getLong("scale"))
                            .setRowCount(table.getRowCount())
                            .setSampleCount(resultSet.getLong("statisticSampleCount"))
                            .setSampleRate(null)
                            .setNullCount(resultSet.getLong("statisticNullCount"))
                            .setDistinctCount(resultSet.getLong("statisticDistinctCount"))
                            .setDensity(resultSet.getBigDecimal("statisticDensity"))
                            .setMinValue(resultSet.getString("statisticMinValue"))
                            .setMaxValue(resultSet.getString("statisticMaxValue"))
                            .setAvgDataSizeByte(resultSet.getBigDecimal("statisticAvgDataSizeByte"))
                            .setEstimateMegaBytes(null)
                            .setStatisticDt(resultSet.getDate("statisticTime"))
                            .setDorisCatalogName(table.getDorisCatalogName())
                            .setSystemCode(table.getSystemCode())
                            .setEnabled(true)
                            .setTransTableName(null)
                            .setTransColumnName(null)
                            .setTransColumnNameManual(false)
                            .setTransComment(null)
                            .setTransCommentManual(false)
                            .setTransDataType(null)
                            .setTransLength(null)
                            .setTransPrecision(null)
                            .setTransScale(null)
                            .setTransPosition(null)
                            .setTransToChar(null)
                            .setTransWarn(null)
                            .setTransError(null);

                    // 计算采样率
                    calcuteSampleRate(column);
                    // 计算列占用空间 setEstimateMegaBytes
                    calcuteEstimateMegaBytes(column);
                    return column;
                });


        columnService.compareData(database, columns);
    }

    private void calcuteSampleRate(SourceColumn column) {
        Long rowCount = column.getRowCount();
        Long sampleCount = column.getSampleCount();
        if (rowCount == null || sampleCount == null || sampleCount == 0) {
            return;
        }
        BigDecimal sampleRate = BigDecimal.valueOf(rowCount).divide(BigDecimal.valueOf(sampleCount), 4, RoundingMode.HALF_UP);
        column.setSampleRate(sampleRate);
    }

    private void calcuteEstimateMegaBytes(SourceColumn column) {
        Long rowCount = column.getRowCount();
        BigDecimal avgDataSizeByte = column.getAvgDataSizeByte();
        if (rowCount == null || avgDataSizeByte == null) {
            return;
        }
        BigDecimal estimateMegaBytes = BigDecimal.valueOf(rowCount).multiply(avgDataSizeByte).divide(BigDecimal.valueOf(1048576), 2, RoundingMode.HALF_UP);
        column.setEstimateMegaBytes(estimateMegaBytes);
    }

    @Transactional
    public void compareData(SourceDatabase database, List<SourceColumn> dealList) {
        List<SourceColumn> originList = select.columns(database);
        // 已存在
        Map<String, SourceColumn> originMap = Maps.uniqueIndex(originList, origin -> origin.columnKey().toString());
        // 待处理
        Map<String, SourceColumn> dealMap = Maps.uniqueIndex(dealList, deal -> deal.columnKey().toString());

        // 新增
        List<SourceColumn> extraList = dealList.stream().filter(deal -> !originMap.containsKey(deal.columnKey().toString())).toList();

        // 保存
        List<SourceColumn> existsList = originList.stream()
                .filter(origin -> dealMap.containsKey(origin.columnKey().toString()))
                .peek(origin -> {
                    SourceColumn deal = dealMap.get(origin.columnKey().toString());
                    origin.setComment(deal.getComment())
                            .setPosition(deal.getPosition())
                            .setDataType(deal.getDataType())
                            .setLength(deal.getLength())
                            .setPrecision(deal.getPrecision())
                            .setScale(deal.getScale())
                            .setRowCount(deal.getRowCount())
                            .setSampleCount(deal.getSampleCount())
                            .setSampleRate(deal.getSampleRate())
                            .setNullCount(deal.getNullCount())
                            .setDistinctCount(deal.getDistinctCount())
                            .setDensity(deal.getDensity())
                            .setMinValue(deal.getMinValue())
                            .setMaxValue(deal.getMaxValue())
                            .setAvgDataSizeByte(deal.getAvgDataSizeByte())
                            .setEstimateMegaBytes(deal.getEstimateMegaBytes())
                            .setStatisticDt(deal.getStatisticDt());
                }).toList();

        // 移除
        List<SourceColumn> missList = originList.stream().filter(item -> !dealMap.containsKey(item.columnKey().toString())).toList();

        add.columns(extraList);
        save.columns(existsList);
        delete.columns(missList);

        log.info("[SYNC] {} -> {} -> 元数据采集 -> SourceColumn             => 共计{} | 新增{} | 删除{}", database.getCatalogName(), database.getDatabaseName(), String.format("%5d", dealList.size()), String.format("%5d", extraList.size()), String.format("%5d", missList.size()) + (
                !missList.isEmpty() ? " => " + missList
                        .stream()
                        .map(column -> column.getTableName() + " - " + column.getColumnName())
                        .collect(Collectors.joining(",")) : ""));
    }
}